package cn.getech.data.development.controller.offline;

import cn.getech.data.development.config.properties.DataDevelopmentConfig;
import cn.getech.data.development.constant.*;
import cn.getech.data.development.entity.TableFieldInfo;
import cn.getech.data.development.entity.TableInfo;
import cn.getech.data.development.model.dto.TableInfoAndFieldDto;
import cn.getech.data.development.model.dto.TableSqlDto;
import cn.getech.data.development.utils.HiveTableUtil;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.R;
import cn.hutool.core.collection.CollectionUtil;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
/**
 * @description: hive表操作类
 * @author: wangzhaowen：kiss
 * @create: 2021/1/15 20:35
 * @version: 2021/1/15
 **/
@Slf4j
public class HiveTableDdl {
    public static R createHiveTable(TableInfo tableInfo, DataDevelopmentConfig hiveConfig) {

        List<String> cols = new ArrayList<>();
        List<String> pars = new ArrayList<>();
        String tableName = tableInfo.getTableName();
        String format = tableInfo.getFormat();
        String dbName = tableInfo.getDbName();
        List<String> kuduKeys = new ArrayList<>();
        List<String> kuduPars = new ArrayList<>();
        List<String> kuduCols = new ArrayList<>();
        List<TableFieldInfo> listFields = tableInfo.getTableFieldInfos();
        listFields.sort(Comparator.comparingInt(TableFieldInfo::getFieldOrder));
        for (TableFieldInfo fieldInfo : listFields) {
            //Hive 中，可以用String、Date和Timestamp表示日期时间
            //String 用 yyyy-MM-dd 的形式表示  Date 用 yyyy-MM-dd 的形式表示  Timestamp 用 yyyy-MM-dd hh:mm:ss 的形式表示
            //所以没有datetime类型。date -> timestamp  ,  timestamp -> string
            String comments=" ";
            if (StringUtils.isNotEmpty(fieldInfo.getFieldAlias())) {
                  comments=" COMMENT '"+fieldInfo.getFieldAlias()+"' ";
            }
            if (TableDataType.DATETIME.getName().equals(fieldInfo.getFieldType())) {
                //Error while compiling statement: FAILED: SemanticException [Error 10099]: DATETIME type isn't supported yet. Please use DATE or TIMESTAMP instead
                kuduCols.add(fieldInfo.getFieldName() + " " + TableDataType.TIMESTAMP.getName());
                if (fieldInfo.getIsPartition() == 0) {
                    cols.add(fieldInfo.getFieldName() + " " + TableDataType.TIMESTAMP.getName()+comments);
                } else {
                    pars.add(fieldInfo.getFieldName() + " " + TableDataType.TIMESTAMP.getName());
                }
            } else if (TableDataType.TIMESTAMP.getName().equals(fieldInfo.getFieldType())) {
                kuduCols.add(fieldInfo.getFieldName() + " " + TableDataType.TIMESTAMP.getName());
                if (fieldInfo.getIsPartition() == 0) {
                    cols.add(fieldInfo.getFieldName() + " " + TableDataType.STRING.getName()+comments);
                } else {
                    pars.add(fieldInfo.getFieldName() + " " + TableDataType.STRING.getName());
                }
            } else if ("varchar".equalsIgnoreCase(fieldInfo.getFieldType())) {
                kuduCols.add(fieldInfo.getFieldName() + " " + TableDataType.TIMESTAMP.getName());
                if (fieldInfo.getIsPartition() == 0) {
                    cols.add(fieldInfo.getFieldName() + " " + TableDataType.STRING.getName()+comments);
                } else {
                    pars.add(fieldInfo.getFieldName() + " " + TableDataType.STRING.getName());
                }
            } else {
                kuduCols.add(fieldInfo.getFieldName() + " " + fieldInfo.getFieldType());
                if (fieldInfo.getIsPartition() == 0) {
                    cols.add(fieldInfo.getFieldName() + " " + fieldInfo.getFieldType()+comments);
                } else {
                    pars.add(fieldInfo.getFieldName() + " " + fieldInfo.getFieldType());
                }
            }
            if (YesOrNoEnum.YES.getValue().equals(fieldInfo.getPrimaryKey())) {
                kuduKeys.add(fieldInfo.getFieldName());
                if (YesOrNoEnum.YES.getValue().equals(fieldInfo.getIsPartition())) {
                    kuduPars.add(fieldInfo.getFieldName());
                }
            }
        }
        String partitions = "";
        if (pars.size() > 0) {
            partitions = String.format("partitioned by(%s)", StringUtils.join(pars.toArray(), ","));
        }
        String kuduPartitions = "";
        if (kuduPars.size() > 0) {
            kuduPartitions = String.format(" PARTITION BY HASH (%s) PARTITIONS 2 \n",
                    StringUtils.join(kuduPars.toArray(), ","));
        }
        String statment = "";
        String jdbcUrl = hiveConfig.getHiveserverurl();
        if (StorageEngineEnum.KUDU.getCode().equals(tableInfo.getStoreEngine())) {
            jdbcUrl = hiveConfig.getKuduJdbcUrl();
            if (CollectionUtils.isEmpty(kuduKeys)) {
                return R.error("kudu 表至少需要配置一个主键字段");
            }
            if (kuduKeys.size() > 1) {
                return R.error("kudu 表只能配置一个主键,请去除多余的主键字段");
            }
            statment = String.format("create  table IF NOT EXISTS %s (\n" +
                            "%s \n" +
                            ", PRIMARY KEY  (%s) \n" +
                            ")\n" +
                            "%s \n" +
                            "STORED AS KUDU \n" +
                            "TBLPROPERTIES (\n" +
                            "'kudu.master_addresses' = '" + hiveConfig.getKuduAddresses() + "'\n" +
                            ")",
                    dbName + "." + tableName,
                    StringUtils.join(kuduCols.toArray(), ","),
                    StringUtils.join(kuduKeys.toArray(), ","),
                    kuduPartitions);
        } else {
            if (DataFormat.text.getName().equals(format)) {
                DelimitFormat delimitFormat;
                if (null != tableInfo.getDelimitId()) {
                    delimitFormat = DelimitFormat.ObjOf(tableInfo.getDelimitId());
                } else {
                    return R.error(DataDevelopmentBizExceptionEnum.NO_DELIMIT.getMessage());
                }
                if (null == delimitFormat) {
                    return R.error(DataDevelopmentBizExceptionEnum.DELIMIT_NOT_EXITS.getMessage());
                }
                String express = delimitFormat.getExpress();
                statment = String.format("create EXTERNAL table IF NOT EXISTS %s (\n" +
                        "%s)\n" +
                        "%s\n" +
                        "row format delimited fields terminated by '%s'\n" +
                        "collection items terminated by ','\n" +
                        "map keys terminated by ':'\n" +
                        "LINES TERMINATED BY '\\n'", tableName, StringUtils.join(cols.toArray(), ","), partitions, express);
            }
            if (DataFormat.json.getName().equals(format)) {
                statment = String.format("create EXTERNAL table IF NOT EXISTS %s (\n" +
                        "%s)\n" +
                        "%s\n" +
                        "ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'\n" +
                        "STORED AS TEXTFILE", tableName, StringUtils.join(cols.toArray(), ","), partitions);
            }
            if (DataFormat.parquet.getName().equals(format)) {
                for (String col : cols) {
                    if (col.split(" ")[1].contains("DATE")) {
                        R.error(DataDevelopmentBizExceptionEnum.DATE_TYPE_NOT_SUPPORT.getMessage());
                    }
                }
                statment = String.format("create EXTERNAL table IF NOT EXISTS %s (\n" +
                        "%s)\n" +
                        "%s\n" +
                        "row format delimited fields terminated by ','\n" +
                        "collection items terminated by ','\n" +
                        "map keys terminated by ':'\n" +
                        "stored as parquet", tableName, StringUtils.join(cols.toArray(), ","), partitions);
            }
        }
        log.info("建表语句为:\n{}",statment);
        HiveTableUtil hiveTableUtil = new HiveTableUtil(jdbcUrl, dbName, "hdfs", "hdfs");
        try {
            hiveTableUtil.execute(statment);
        } catch (SQLException e) {
            if (e.getMessage() != null) {
                if (e.getMessage().contains("Kudu PRIMARY KEY columns must be specified as the first columns in the table")) {
                    return  R.error("Kudu 表的主键字段必须放在第一个");
                }
            }
            if (e.getMessage() != null && e.getMessage().contains("primary key")) {
                return  R.error(DataDevelopmentBizExceptionEnum.PRIMARY_KEY_EXIST.getMessage());
            } else {
                return  R.error(DataDevelopmentBizExceptionEnum.CREATE_TABLE_ERROR.getMessage()+",详情如下:<br/>"+e.getMessage());
            }
        } catch (Exception e) {
                return  R.error(DataDevelopmentBizExceptionEnum.CREATE_TABLE_ERROR.getMessage()+",详情如下:<br/>"+e.getMessage());
        }finally {
            hiveTableUtil.close();
        }
        return R.ok();
    }

    public static Boolean updateKuduTable(List<TableInfoAndFieldDto> needUpdates, TableInfo tableInfo, DataDevelopmentConfig hiveConfig) {
        Boolean falge = true;
        String jdbcUrl = hiveConfig.getKuduJdbcUrl();
        HiveTableUtil hiveTableUtil = new HiveTableUtil(jdbcUrl, tableInfo.getDbName(), "hdfs", "hdfs");
        if(CollectionUtil.isNotEmpty(needUpdates)){
            for (TableInfoAndFieldDto needUpdate : needUpdates) {
                List<TableSqlDto> tableSqlDtos = needUpdate.getTableSqlDtos();
                if(CollectionUtil.isNotEmpty(tableSqlDtos)){
                    for (TableSqlDto tableSqlDto : tableSqlDtos) {
                        try {
                            //是否执行
                            tableSqlDto.setIsExces(1);
                            hiveTableUtil.execute(tableSqlDto.getSql());
                        } catch (Exception e){
                            falge = false;
                            log.error("修改kudu表结构异常:\n表:{},error:{}",tableInfo.getTableName(),e.getMessage());
                            tableSqlDto.setIsSuccess(1);
                            tableSqlDto.setMsg(e.getMessage());
                            //关流
                            hiveTableUtil.close();
                            return falge;
                        }
                    }
                }
            }
        }
        //关流
        if(null != hiveTableUtil){
            hiveTableUtil.close();
        }
        return falge;
    }

    public static void rockKuduSqls(List<TableSqlDto> needRockSqls,
                                    TableInfo tableInfo, DataDevelopmentConfig hiveConfig) {
        String jdbcUrl = hiveConfig.getKuduJdbcUrl();
        HiveTableUtil hiveTableUtil = new HiveTableUtil(jdbcUrl, tableInfo.getDbName(), "hdfs", "hdfs");
        if(CollectionUtil.isNotEmpty(needRockSqls)){
            for (TableSqlDto needRockSql : needRockSqls) {
                try {
                    hiveTableUtil.execute(needRockSql.getSql());
                }catch (Exception e){
                    log.error("修改kudu表异常并进行回滚也异常:\n表:{},error:{}",tableInfo.getTableName(),e.getMessage());
                }
            }
        }
        if(null != hiveTableUtil){
            hiveTableUtil.close();
        }
    }
}
